#include "rpcbalancer.h"

BackendSession::BackendSession(EventLoop *loop, const InetAddress &serverAddr)
    : _loop(loop),
      _client(loop, serverAddr),
      _codec(std::bind(&BackendSession::onRpcMessage, this, _1, _2, _3),
             std::bind(&BackendSession::onRawMessage, this, _1, _2, _3)),
      _nextId(0)
{
    _client.setConnectionCallback(
        std::bind(&BackendSession::onConnection, this, _1));
    _client.setMessageCallback(
        std::bind(&rpc::RpcCodec::onMessage, &_codec, _1, _2, _3));
    _client.enableRetry();
}

void BackendSession::onConnection(const TcpConnectionPtr &conn)
{
    _loop->assertInLoopThread();
    LOG_INFO << "Backend "
             << conn->localhostAddress().toHostPort() << " -> "
             << conn->peerAddress().toHostPort() << " is "
             << (conn->connected() ? "UP" : "DOWN");
    if (conn->connected())
    {
        _conn = conn;
    }
    else
    {
        _conn.reset();
    }
}
void BackendSession::onRpcMessage(const TcpConnectionPtr &conn,
                                  const rpc::RpcMessagePtr msg,
                                  Timestamp)
{
    onMessageT(*msg);
}

bool BackendSession::onRawMessage(const TcpConnectionPtr &,
                                  std::string message,
                                  Timestamp)
{
    RawMessage raw(message);
    if (raw.parse(_codec.tag())) //注意解析失败才发送
    {
        onMessageT(raw);
        return false;
    }
    else
    {
        return true;
    }
}
//=============================================================

Balancer::Balancer(EventLoop *loop,
                   const InetAddress &listenAddr,
                   const std::vector<InetAddress> &backends)
    : _server(loop, listenAddr),
      _codec(std::bind(&Balancer::onRpcMessage, this, _1, _2, _3),
             std::bind(&Balancer::onRawMessage, this, _1, _2, _3)),
      _backends(backends)
{
    _server.setThreadInitCallback(std::bind(&Balancer::initPerThread, this, _1));
    _server.setConnectionCallback(std::bind(&Balancer::onConnection, this, _1));
    _server.setMessageCallback(std::bind(&rpc::RpcCodec::onMessage, &_codec, _1, _2, _3));
}

void Balancer::initPerThread(EventLoop *ioLoop)
{
    int count = _threadCount.getAndAdd(1); //多个线程共用
    LOG_INFO << "IO thread " << count;
    _t_backends::instance().current = count % _backends.size(); //代表分派到哪个主服务器
    for (size_t i = 0; i < _backends.size(); ++i)
    {
        _t_backends::instance().backends.emplace_back(
            new BackendSession(ioLoop, _backends[i])); //一个EventLoop per thread 连接N个主服务器
        _t_backends::instance().backends.back()->connect();
    }
}

void Balancer::onConnection(const TcpConnectionPtr &conn)
{
    LOG_INFO << "Client "
             << conn->peerAddress().toHostPort() << " -> "
             << conn->localhostAddress().toHostPort() << " is "
             << (conn->connected() ? "UP" : "DOWN");
    if (!conn->connected())
    {
        // FIXME: cancel outstanding calls, otherwise, memory leak
    }
}

void Balancer::onRpcMessage(const TcpConnectionPtr &conn,
                            const rpc::RpcMessagePtr &msg,
                            Timestamp)
{
    onMessageT(conn, *msg);
}
bool Balancer::onRawMessage(const TcpConnectionPtr &conn,
                            std::string message,
                            Timestamp)
{
    RawMessage raw(message);
    if (raw.parse(_codec.tag()))
    {
        onMessageT(conn, raw);
        return false;
    }
    else
        return true; // try normal rpc message callback
}

bool RawMessage::parse(const std::string &tag)
{
    const char *const body = message.data() + ProtobufCodecLite::kHeaderLen;//M+N+4(首地址的char*)
    const int bodylen = message.size() - ProtobufCodecLite::kHeaderLen; // M+N+4;
    const int taglen = static_cast<int>(tag.size());                    // M
    //验证
    if (ProtobufCodecLite::validateChecksum(body, bodylen) && // 1.校验和
        (memcmp(body, tag.data(), tag.size()) == 0) &&        // 2.type
        (bodylen >= taglen + 3 + 8))                          // 3."RPC0"(4) + 3 + cheksum(4)
    {
        const char *const p = body + taglen;
        uint8_t type = *(p + 1);
        //p[0] == 0x08
        //p[1] == 0x01 || 0x02 
        //p[2] == 0x11
        if (*p == 0x08 && (type == 0x01 || type == 0x02) && *(p + 2) == 0x11)
        {
            uint64_t x = 0;
            memcpy(&x, p + 3, sizeof(x));//set后面的ID
            set_id(le64toh(x));//le*toh表示小端到主机
            _loc = p + 3;
            return true;
        }
    }
    return false;
}
void RawMessage::updateId()
{
    uint64_t le64 = htole64(_id);
    memcpy(const_cast<void *>(_loc), &le64, sizeof(le64));
    //message的id已经更新过了
    const char *body = message.data() + ProtobufCodecLite::kHeaderLen;
    int bodylen = message.size() - ProtobufCodecLite::kHeaderLen;
    
    int32_t checkSum = ProtobufCodecLite::checksum(body, bodylen - ProtobufCodecLite::kChecksumLen);
    
    int32_t be32 = sockets::hostToNetwork32(checkSum);
    memcpy(const_cast<char *>(body + bodylen - ProtobufCodecLite::kChecksumLen), &be32, sizeof(be32));
}